#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <queue>
#include <semaphore>
#include <thread>
#include <unordered_map>
#include <vector>
using ThreadFunc = std::function<void(int)>;

class Result;
class Any;
class Task {
 public:
  Task() : result_(nullptr) {}
  ~Task() = default;
  virtual Any Run() = 0;
  void SetResult(Result* result) {
    if (result != nullptr) result_ = result;
  }
  void Exec();

 private:
  // std::shared_ptr<Result> result_;造成循环引用问题
  // TODO:使用weak_ptr
  Result* result_;
  // Result类生命周期比Task长
};

// -----------------------------------
class Any {
 public:
  Any() = default;
  ~Any() = default;
  Any(const Any&) = delete;
  Any& operator=(const Any&) = delete;
  Any(Any&&) = default;
  Any& operator=(Any&&) = default;

  template <typename T>
  Any(T data) : base_(std::make_unique<Derived<T>>(data)) {}

  template <typename T>
  T Cast() {
    Derived<T>* ptr = dynamic_cast<Derived<T>*>(base_.get());
    if (ptr == nullptr) {
      throw " type is unmatched!";
    }
    return ptr->data_;
  }

 private:
  class Base {
   public:
    virtual ~Base() = default;
  };
  template <typename T>
  class Derived : public Base {
   public:
    Derived(T data) : data_(data) {}
    ~Derived() = default;
    T data_;
  };

  std::unique_ptr<Base> base_;
};

// ------------------
// TODO:是使用的二元信号量吗？
class Semaphore {
 public:
  Semaphore(int rlimit = 0) : rlimit_(rlimit) {}
  ~Semaphore() = default;
  void Wait();
  void Post();

 private:
  int rlimit_;
  std::mutex mutex_;
  std::condition_variable cond_;
};
void Semaphore::Wait() {
  std::unique_lock<std::mutex> lock(mutex_);
  cond_.wait(lock, [&]() -> bool { return rlimit_ > 0; });
  rlimit_--;
}
void Semaphore::Post() {
  std::unique_lock<std::mutex> lock(mutex_);
  rlimit_++;
  cond_.notify_all();
  // cond_.notify_all() TODO:这里用这个还是用那个？
}
// -----------------------
class Result {
 public:
  Result(std::shared_ptr<Task> task, bool flag = true)
      : task_(task), is_valid_(flag) {
    task_->SetResult(this);  // *************************
  }
  ~Result() = default;
  // 因为Any类的拷贝构造和拷贝赋值运算符已删除，所以Result类也只能删除
  Result(const Result&) = delete;
  Result& operator=(const Result&) = delete;
  // Result(Result&&) = default;
  // Result& operator=(Result&&) = default;  // 移动赋值运算符
  // Result&& operator=(const Result&&) = default; 错误

  void Set(Any&& any);
  Any Get();

 private:
  Any any_;
  Semaphore sem_;
  std::atomic_bool is_valid_;

  std::shared_ptr<Task> task_;  // 指向对应获取返回值的任务对象
};

void Result::Set(Any&& any) {
  any_ = std::move(any);
  sem_.Post();
}
Any Result::Get() {
  if (!is_valid_) return nullptr;
  sem_.Wait();
  return std::move(any_);
}

// TODO:线程安全
static std::atomic_int id = 1;
int GenerateId() {
  id++;
  return id;
}
class Thread {
 public:
  Thread(ThreadFunc func) : func_(std::move(func)), id_(GenerateId()) {}
  ~Thread() {}
  void Start();

  int GetId() { return id_; }

 private:
  ThreadFunc func_;
  int id_;
  std::unique_ptr<std::thread> thread_;
};
enum class MODE {
  FIXED,
  CACHED,
};

class ThreadPool {
 public:
  ThreadPool(MODE mode = MODE::FIXED)
      : thread_cur_num_(0),
        thread_idle_num_(0),
        task_num_(0),
        running_(false),
        mode_(mode) {
    task_num_ = 0;
  }
  ~ThreadPool() {
    running_ = false;
    // 等待线程池里面所有的线程返回  有两种状态：阻塞 & 正在执行任务中
    std::unique_lock<std::mutex> lock(mtx_);
    consumer_cond_.notify_all();
    exit_cond_.wait(lock, [&]() -> bool { return thread_map_.size() == 0; });
  }
  ThreadPool(const ThreadPool&) = delete;
  ThreadPool operator=(const ThreadPool&) = delete;

  void SetThreadNum(int thread_init_num = -1);
  void SetMode(MODE mode) { mode_ = mode; }

  void Start();
  Result AddTask(std::shared_ptr<Task> sp);

 private:
  void ThreadRunFunc(int thread_id);
  bool IsRunning() const { return running_; }

 private:
  std::unordered_map<int, std::unique_ptr<Thread>> thread_map_;
  static int thread_init_num_;       // 初始线程数量
  std::atomic_int thread_cur_num_;   //当前线程数量
  std::atomic_int thread_idle_num_;  // 空闲线程数量
  static int thread_map_max_num_;    // 线程数量上限
  static int thread_max_idle_time_;  // 动态线程最长空闲时间
  ThreadFunc func_;

  std::queue<std::shared_ptr<Task>> task_que_;
  std::atomic_int task_num_;  // 任务队列中当前任务数量
  static int task_que_max_num_;

  std::mutex mtx_;
  std::condition_variable producer_cond_;
  std::condition_variable consumer_cond_;
  std::condition_variable exit_cond_;

  std::atomic_bool running_;
  MODE mode_;
};
int ThreadPool::thread_init_num_ = 4;
int ThreadPool::thread_map_max_num_ = 8;
int ThreadPool::task_que_max_num_ = INT32_MAX;
int ThreadPool::thread_max_idle_time_ = 10;

void Thread::Start() {
  thread_ = std::make_unique<std::thread>(func_, id_);
  thread_->detach();
}
// 用户向线程池提交任务,生产！
Result ThreadPool::AddTask(std::shared_ptr<Task> sp) {
  std::unique_lock<std::mutex> ul(mtx_);
  if (!producer_cond_.wait_for(ul, std::chrono::seconds(1), [&]() -> bool {
        return task_num_ < task_que_max_num_;
      })) {
    std::cerr << "task_que is full, AddTask failed!" << std::endl;
    return Result(sp, false);
  }
  task_que_.emplace(sp);  // task_que_.emplace(std::move(sp));
  task_num_++;
  std::cout << "AddTask successfully!" << std::endl;
  // consumer_cond_.notify_all();  // FIXME:是否这行代码不用锁住？
  consumer_cond_.notify_all();

  if (mode_ == MODE::CACHED && task_num_ > thread_idle_num_ &&
      thread_cur_num_ < thread_map_max_num_) {
    // 创建线程、添加进map、Start thread、相关成员变量修改
    auto t = std::make_unique<Thread>(
        std::bind(&ThreadPool::ThreadRunFunc, this, std::placeholders::_1));
    int thread_id = t->GetId();
    thread_map_.insert({thread_id, std::move(t)});
    thread_map_[thread_id]->Start();
    thread_cur_num_++;
    thread_idle_num_++;
    std::cout << "create new thread to do tasks!" << std::endl;
  }

  return Result(sp);
}
// 具体的一个任务执行流程，消费！
void ThreadPool::ThreadRunFunc(int thread_id) {
  auto last = std::chrono::high_resolution_clock().now();
  while (1) {
    std::shared_ptr<Task> task;
    {
      std::unique_lock<std::mutex> ul(mtx_);

      std::cout << "tid:" << std::this_thread::get_id() << "尝试获取任务..."
                << std::endl;

      while (task_num_ == 0) {
        if (!running_) {
          thread_map_.erase(thread_id);  // std::this_thread::getid()
          std::cout << "threadid:" << std::this_thread::get_id() << " exit!"
                    << std::endl;
          exit_cond_.notify_all();
          return;  // 线程函数结束，线程结束
        }

        if (mode_ == MODE::CACHED) {
          if (std::cv_status::timeout ==
              consumer_cond_.wait_for(ul, std::chrono::seconds(1))) {
            auto now = std::chrono::high_resolution_clock::now();
            auto dur =
                std::chrono::duration_cast<std::chrono::seconds>(now - last);
            if (dur.count() >= thread_max_idle_time_ &&
                thread_cur_num_ > thread_init_num_) {
              std::cout << "thread " << thread_id << " 超时被干掉了"
                        << std::endl;
              thread_idle_num_--;
              thread_cur_num_--;
              thread_map_.erase(thread_id);
              return;
            }
          }

        } else
          consumer_cond_.wait(ul);
      }
      thread_idle_num_--;
      std::cout << "tid:" << std::this_thread::get_id() << "获取任务成功..."
                << std::endl;
      task = task_que_.front();
      task_que_.pop();
      task_num_--;
      if (task_num_ > 0) consumer_cond_.notify_one();
      producer_cond_.notify_all();
    }
    if (task != nullptr) {
      task->Exec();
    }
    thread_idle_num_++;
    last = std::chrono::high_resolution_clock().now();
  }
}
void ThreadPool::Start() {
  running_ = true;
  for (int i = 0; i < thread_init_num_; i++) {
    auto t = std::make_unique<Thread>(
        std::bind(&ThreadPool::ThreadRunFunc, this, std::placeholders::_1));
    thread_cur_num_++;
    thread_map_.insert({t->GetId(), std::move(t)});
  }
  for (const auto& t : thread_map_) {
    t.second->Start();
    thread_idle_num_++;
  }
}

void ThreadPool::SetThreadNum(int thread_init_num) {
  if (thread_init_num == -1)
    thread_init_num_ =
        std::thread::hardware_concurrency();  // TODO:1807496520 bug?
  else
    thread_init_num_ = thread_init_num;
}

void Task::Exec() {
  if (result_ != nullptr) result_->Set(Run());
}
